import pandas as pd
from pathlib import Path
import sys, os
from concurrent.futures import ProcessPoolExecutor, as_completed

# ── 根目录选择 ────────────────────────────────────
mac_path = Path("/Users/eddy/Desktop")
win_path = Path("G:/trading_2025")
base_path = win_path if sys.platform.startswith("win") else mac_path

data_path = base_path / "shared_data" / "raw_data" / "binance_data" / "spot" / "daily"

# 输出目录一次性创建
exchange, product = "binance", "spot"
output_root = Path.cwd().parent / "data" / exchange / "full_data"
output_root.mkdir(parents=True, exist_ok=True)


# ── 单个 symbol 的处理逻辑 ────────────────────────
def process_symbol(symbol_dir: Path) -> str:
    symbol = symbol_dir.name
    print(f'processing {symbol}')

    # 读取并合并
    df = pd.concat((pd.read_csv(f) for f in symbol_dir.iterdir()), ignore_index=True)

    # 清洗
    df = df[["datetime", "open", "high", "low", "close", "volume"]]
    df["datetime"] = df["datetime"].str.replace(r"\+.*$", "", regex=True)
    df = df.drop_duplicates(subset="datetime")

    # 保存
    out_file = output_root / f"{symbol}-{product}.csv"
    df.to_csv(out_file, index=False)
    return f"{symbol} → saved {out_file}"


# ── 并发执行 ───────────────────────────────────────
if __name__ == "__main__":            # Windows 必须加
    symbol_dirs = [p for p in data_path.iterdir() if p.is_dir()]

    with ProcessPoolExecutor(max_workers=4) as pool:
        futures = {pool.submit(process_symbol, p): p.name for p in symbol_dirs}

        for fut in as_completed(futures):
            print(fut.result())
